package cn.doitedu.sql;

import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/4/12
 * @Desc: 学大数据，上多易教育
 *
 * kafka中不断地流入如下数据：
 * {"uid":1,"event_id":"page_load","properties":{"url":"/a","ref":"/x"},"action_time":1704719574000}
 * {"uid":1,"event_id":"add_cart","properties":{"url":"/a","item_id":"p01"},"action_time":1704719574000}
 * {"uid":1,"event_id":"page_load","properties":{"url":"/b","ref":"/a"},"action_time":1704719574000}
 * {"uid":2,"event_id":"page_load","properties":{"url":"/a","ref":"/x"},"action_time":1704719574000}
 *
 * {"uid":2,"event_id":"video_play","properties":{"video_id":"v001","play_id":"play001"},"action_time":1704719574000}
 * {"uid":2,"event_id":"video_hb","properties":{"video_id":"v001","play_id":"play001"},"action_time":1704719575000}
 * {"uid":2,"event_id":"video_hb","properties":{"video_id":"v001","play_id":"play001"},"action_time":1704719580000}
 * {"uid":2,"event_id":"video_hb","properties":{"video_id":"v001","play_id":"play001"},"action_time":1704719585000}
 * {"uid":2,"event_id":"video_pause","properties":{"video_id":"v001","play_id":"play001"},"action_time":1704719590000}
 *
 * {"uid":2,"event_id":"video_resume","properties":{"video_id":"v001","play_id":"play001"},"action_time":1704719800000}
 * {"uid":2,"event_id":"video_hb","properties":{"video_id":"v001","play_id":"play001"},"action_time":1704719805000}
 * {"uid":2,"event_id":"video_hb","properties":{"video_id":"v001","play_id":"play001"},"action_time":1704719805000}
 * {"uid":2,"event_id":"video_hb","properties":{"video_id":"v001","play_id":"play001"},"action_time":1704719805000}
 * {"uid":2,"event_id":"video_pause","properties":{"video_id":"v001","play_id":"play001"},"action_time":1704719810000}
 *
 * {"uid":2,"event_id":"video_resume","properties":{"video_id":"v001","play_id":"play001"},"action_time":1704720000000}
 * {"uid":2,"event_id":"video_hb","properties":{"video_id":"v001","play_id":"play001"},"action_time":1704720005000}
 * {"uid":2,"event_id":"video_stop","properties":{"video_id":"v001","play_id":"play001"},"action_time":1704720007000}
 *
 *
 * {"uid":2,"event_id":"page_load","properties":{"url":"/b","ref":"/x"},"action_time":1704719574000}
 * {"uid":1,"event_id":"page_load","properties":{"url":"/c","ref":"/x"},"action_time":1704719574000}
 * {"uid":1,"event_id":"item_share","properties":{"url":"/a","item_id":"p03"},"action_time":1704719574000}
 *
 *
 * 实时计算： 每个视频的每次播放的总时长,结果写入mysql
 *
 * 抽象：
 * 1,v01,play01, play ,t1     ,0
 * 1,v01,play01, hb ,t2       ,0
 * 1,v01,play01, hb ,t3       ,0
 * 1,v01,play01, pause ,t4    ,0
 *
 * 1,v01,play01, resume ,t8   ,1
 * 1,v01,play01, hb ,t9       ,0
 * 1,v01,play01, hb ,t10      ,0
 * 1,v01,play01, stop ,t11    ,0
 *
 **/
public class Demo03_Exercise {

    public static void main(String[] args) {


        // 构造编程入口环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.getCheckpointConfig().setCheckpointTimeout(5000);
        env.setParallelism(1);

        env.setStateBackend(new HashMapStateBackend());

        // 创建表环境
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        tenv.executeSql(
                "create table user_events_kafka(\n" +
                        "     uid bigint,\n" +
                        "     event_id string,\n" +
                        "     properties map<string,string>,\n" +
                        "     action_time bigint ,  \n" +
                        "     rt as to_timestamp_ltz(action_time,3) ,  \n" +
                        "     watermark for rt as rt - interval '0' second  \n" +
                        ") with (\n" +
                        "  'connector' = 'kafka',\n" +
                        "  'topic' = 'tpc-a',\n" +
                        "  'properties.bootstrap.servers' = 'doitedu:9092',\n" +
                        "  'properties.group.id' = 'doit47-g2',\n" +
                        "  'scan.startup.mode' = 'latest-offset',\n" +
                        "  'value.format' = 'json',\n" +
                        "  'value.fields-include' = 'EXCEPT_KEY'\n" +
                        ")");

        //tenv.executeSql("select * from user_events_kafka order by rt").print();
        //System.exit(1);




        tenv.executeSql("CREATE TABLE play_timelong_kafka      ( \n" +
                "  video_id STRING,                                \n" +
                "  play_id  STRING,                                \n" +
                "  time_long  BIGINT,                                \n" +
                "  primary key(video_id,play_id) not enforced  \n" +
                ") WITH (                                     \n" +
                "  'connector' = 'jdbc',                     \n" +
                "  'url' = 'jdbc:mysql://doitedu:3306/doit47', \n" +
                "  'table-name' = 'play_timelong',              \n" +
                "  'username' = 'root',                      \n" +
                "  'password' = 'root'                       \n" +
                ")");

        tenv.executeSql(
                "WITH tmp1 AS (\n" +
                        "    SELECT\n" +
                        "        properties['video_id'] as video_id,\n" +
                        "        properties['play_id'] as play_id,\n" +
                        "        event_id,\n" +
                        "        action_time,\n" +
                        "        rt,\n" +
                        "        if(event_id = 'video_resume',1,0) as flag\n" +
                        "    from user_events_kafka\n" +
                        "    where event_id in ('video_play','video_hb','video_pause','video_resume','video_stop') \n" +
                        ")\n" +
                        ",tmp2 AS (\n" +
                        "SELECT\n" +
                        "        video_id,\n" +
                        "        play_id,\n" +
                        "        event_id,\n" +
                        "        action_time,\n" +
                        "        sum(flag) over(partition by video_id,play_id order by rt ) as flag\n" +
                        "FROM  tmp1\n" +
                        ")\n" +
                        ",tmp3 AS (\n" +
                        "    SELECT\n" +
                        "    video_id,\n" +
                        "    play_id,\n" +
                        "    flag,\n" +
                        "    max(action_time) - min(action_time) as part_timelong \n" +
                        "    from tmp2 \n" +
                        "    group by video_id,play_id,flag\n" +
                        ")\n" +
                        "\n" +
                        "SELECT\n" +
                        "\n" +
                        "video_id,\n" +
                        "play_id,\n" +
                        "sum(part_timelong) as time_long \n" +
                        "\n" +
                        "from tmp3\n" +
                        "group by video_id,play_id").print();



    }

}
